"""
本地文件系统的资源模型
"""
import os
import shutil
import stat
import xml.etree.ElementTree as ET
from email.utils import formatdate
from os import stat_result
from pathlib import PurePath
from time import timezone, localtime, strftime
from typing import List, Union, Tuple, Literal, Set, AsyncGenerator, Optional, Dict
from urllib.parse import urlparse, unquote

import anyio
from fastapi import Depends, HTTPException, Request, Header, Response
from fastapi.responses import StreamingResponse
from starlette._compat import md5_hexdigest
from starlette.responses import guess_type

from my_webdav.config import app_setting
from my_webdav.resource_model.base_model import DavCollection, DavFileMember, ResourceCreator

LFSModel = Union['LFSCollection', 'LFSFileMember']
ResourceMode = Literal[1, 2]

BASIC_PROPS: Set[str] = {'displayname', 'getcontenttype', 'getcontentlength', 'creationdate', 'iscollection'}
ALL_PROPS: Set[str] = {
    'displayname', 'getcontenttype', 'getcontentlength', 'creationdate', 'iscollection', 'ishidden', 'isreadonly',
    'resourcetype'
}


def convert_to_local_path(path: PurePath) -> PurePath:
    rel_path = path.relative_to(app_setting.LOCAL_MOUNT_PATH)
    if rel_path.as_posix().startswith(".."):
        raise ValueError
    local_path = app_setting.LOCAL_ROOT_PATH.joinpath(rel_path)
    return local_path


def request_path_to_local(request: Request) -> PurePath:
    """
    将请求路径转为对应的本地文件路径
    以PurePath对象做路径计算 避免IO操作
    :param request:
    :return:
    """
    try:
        req_path = PurePath(request.scope['path'])
        local_path = convert_to_local_path(req_path)
    except Exception:
        raise HTTPException(404)

    return local_path


def dst_path_to_local(dst: Optional[str] = Header(alias='Destination', default=None)) -> PurePath:
    try:
        url_path = PurePath(urlparse(unquote(dst)).path)
        local_path = convert_to_local_path(url_path)
    except Exception:
        raise HTTPException(400)
    return local_path


async def get_os_stat(local_path: PurePath = Depends(request_path_to_local)) -> Tuple[os.stat_result, PurePath]:
    """
    若路径存在则返回对应的os.stat和本地路径, 否则返回404
    :param local_path:
    :return:
    """
    try:
        stat_res = await anyio.to_thread.run_sync(os.stat, local_path)
        return stat_res, local_path
    except FileNotFoundError:
        raise HTTPException(status_code=404, detail="Not Found")


def local_model_adapter(
        info: Tuple[stat_result, PurePath] = Depends(get_os_stat)
) -> LFSModel:
    stat_res, path = info
    mode = stat_res.st_mode
    if stat.S_ISREG(mode):
        return LFSFileMember(path, stat_res)
    elif stat.S_ISDIR(mode):
        return LFSCollection(path, stat_res)
    else:
        # 非常规文件和目录
        raise HTTPException(status_code=422)


def get_resource_creator(local_path: PurePath = Depends(request_path_to_local)) -> 'LocalResourceCreator':
    creator = LocalResourceCreator(local_path)
    return creator


def get_href(local_path: PurePath, flg: ResourceMode, mnt_path: PurePath, root_path: PurePath) -> str:
    rel_path = local_path.relative_to(root_path)  # 相对于本地文件挂载目录的路径
    tail = '/' if flg == 1 else ''
    href = mnt_path.joinpath(rel_path).as_posix()  # 挂载路径+相对路径
    if href != '/':
        href = f"{href}{tail}"
    return href


def get_displayname(href: str) -> str:
    if href == "/":
        return "/"
    else:
        return PurePath(href).name


def get_creation_date(s) -> str:
    tz = format(int(timezone / 3600), "+03d")
    c_time = f"{strftime('%Y-%m-%dT%H:%M:%S', localtime(s))}{tz}:00"
    return c_time


async def walk_dir(
        root_path, depth: Literal["1", "infinity"], topdown: bool = True
) -> AsyncGenerator[LFSModel, None]:
    """
    遍历root_path子成员, yield对应的资源模型. 不包括root_path
    :param topdown:
    :param root_path:
    :param depth:
    :return:
    """
    recursion_depth = 0
    depth_ = 1 if depth == "1" else None
    for root, dirs, files in await anyio.to_thread.run_sync(os.walk, root_path, topdown):
        for name in (dirs + files):
            info = await get_os_stat(PurePath(root).joinpath(name))
            res_model = local_model_adapter(info)
            yield res_model

        recursion_depth += 1
        if recursion_depth == depth_:
            break


async def target_exist(dst: PurePath) -> Optional[LFSModel]:
    """
    dst对应的资源存在则返回相应的LFSModel, 否则返回None
    :param dst:
    :return:
    """
    try:
        info = await get_os_stat(dst)
    except HTTPException:
        return None

    model = local_model_adapter(info)
    return model


class LFSMember(object):
    FLG: ResourceMode

    def __init__(self, local_path: PurePath, stat_res: stat_result):
        self.local_path = local_path
        self.f_name = local_path.name
        self.stat_res = stat_res
        self.href = get_href(self.local_path, self.FLG, app_setting.LOCAL_MOUNT_PATH, app_setting.LOCAL_ROOT_PATH)
        self.displayname = get_displayname(self.href)

        self.PROP_MAPPING: Dict[str, Optional[str]] = {
            "getlastmodified": formatdate(self.stat_res.st_mtime, usegmt=True),
            "displayname": self.displayname,
            "getetag": md5_hexdigest(
                f"{self.stat_res.st_mtime}-{self.stat_res.st_size}".encode(), usedforsecurity=False
            ),
            "getcontenttype": "httpd/unix-directory" if self.FLG == 1 else guess_type(self.f_name)[0],
            "iscollection": '1' if self.FLG == 1 else '',
            "getcontentlength": '' if self.FLG == 1 else str(self.stat_res.st_size),
            "ishidden": "1" if self.f_name[0] == "." else '',
            "isroot": "1" if PurePath(self.href).as_posix() == app_setting.LOCAL_MOUNT_PATH else '',
            "href": self.href,
            "creationdate": get_creation_date(self.stat_res.st_ctime)
        }

        self.size = None if self.FLG == 1 else self.stat_res.st_size

    async def prop_xml(
            self, depth: str, props: List[str], only_propname: bool = False, all_prop: bool = False
    ) -> bytes:
        if only_propname or all_prop:
            props = list(ALL_PROPS)

        multi_status_elem = ET.Element("D:multistatus", {"xmlns:D": "DAV:"})
        resp = await self.compose_resp_xml(props, only_propname)
        multi_status_elem.append(resp)

        if depth != "0":
            depth: Literal["1", "infinity"]
            async for res_model in walk_dir(self.local_path, depth):
                resp = await res_model.compose_resp_xml(props, only_propname)
                multi_status_elem.append(resp)

        return ET.tostring(multi_status_elem, xml_declaration=True, encoding='utf-8')

    async def get_prop(self, prop: str) -> ET.Element:
        self: LFSModel
        elem = ET.Element(f"D:{prop}")

        prop_val = self.PROP_MAPPING.get(prop.lower())
        if prop_val:
            elem.text = prop_val
            return elem

        if prop.lower() == "isreadonly":
            writeable = await anyio.to_thread.run_sync(os.access, self.local_path, os.W_OK)
            if not writeable:
                elem.text = "1"
            return elem

        if prop.lower() == "resourcetype":
            if self.FLG == 1:
                elem.append(ET.Element("D:collection"))

        return elem

    async def compose_resp_xml(self, props: List[str], only_propname: bool = False) -> ET.Element:
        """
        组装单个资源响应的xml
        :param only_propname: 只返回属性名
        :param props:
        :return:
        """
        resp_root = ET.Element("D:response")

        href_elem = await self.get_prop("href")
        prop_stat_elem = ET.Element("D:propstat")
        prop_elem = ET.Element("D:prop")

        for prop in props:
            if only_propname:
                elem = ET.Element(f"D:{prop}")
            else:
                elem = await self.get_prop(prop)
            prop_elem.append(elem)

        status_elem = ET.Element("D:status")
        status_elem.text = "HTTP/1.1 200 OK"

        prop_stat_elem.append(prop_elem)
        prop_stat_elem.append(status_elem)

        resp_root.append(href_elem)
        resp_root.append(prop_stat_elem)

        return resp_root

    async def delete(self):
        pass

    # 暂未实现Lock相关的功能
    async def lock(self):
        pass

    async def unlock(self):
        pass

    async def move(self, dst: Optional[str], overwrite: Literal["F", "T"]):
        pass

    async def copy(
            self, dst: Optional[str], overwrite: Literal["F", "T"], depth: Literal["0", "infinity"]
    ):
        pass


class LFSCollection(DavCollection, LFSMember):

    async def delete(self):
        try:
            await anyio.to_thread.run_sync(shutil.rmtree, self.local_path)
        except Exception:
            raise HTTPException(423)

    async def move(self, dst: Optional[str], overwrite: Literal["F", "T"]):
        dst = dst_path_to_local(dst)
        dst_model = await target_exist(dst)
        if dst_model is None:
            await anyio.to_thread.run_sync(os.rename, self.local_path, dst)
        else:
            # 目标资源存在
            if overwrite == 'T':
                await dst_model.delete()
                await anyio.to_thread.run_sync(os.rename, self.local_path, dst)
            else:
                # RFC-4918 9.9.4
                raise HTTPException(412)

    async def copy(
            self, dst: Optional[str], overwrite: Literal["F", "T"], depth: Literal["0", "infinity"]
    ):
        dst = dst_path_to_local(dst)
        if dst == self.local_path:
            raise HTTPException(403)

        dst_model = await target_exist(dst)
        if dst_model is None:
            if depth == '0':
                # 新建目录
                await anyio.to_thread.run_sync(os.mkdir, dst)
            else:
                # 复制目录
                await anyio.to_thread.run_sync(shutil.copytree, self.local_path, dst)
        else:
            # 目标处存在资源
            if overwrite == 'F':
                raise HTTPException(412)  # RFC-4918 9.8.5
            else:
                await dst_model.delete()
                await anyio.to_thread.run_sync(shutil.copytree, self.local_path, dst)


class LFSFileMember(DavFileMember, LFSMember):

    async def delete(self):
        try:
            await anyio.to_thread.run_sync(os.remove, self.local_path)
        except Exception:
            raise HTTPException(423)

    async def move(self, dst: Optional[str], overwrite: Literal["F", "T"]):
        dst = dst_path_to_local(dst)
        dst_model = await target_exist(dst)
        if dst_model is None:
            await anyio.to_thread.run_sync(os.rename, self.local_path, dst)
        else:
            # 目标资源存在
            if overwrite == 'T':
                await dst_model.delete()
                await anyio.to_thread.run_sync(os.rename, self.local_path, dst)
            else:
                # RFC-4918 9.9.4
                raise HTTPException(412)

    async def copy(
            self, dst: Optional[str], overwrite: Literal["F", "T"], depth: Literal["0", "infinity"]
    ):
        dst = dst_path_to_local(dst)
        if dst == self.local_path:
            raise HTTPException(403)

        dst_model = await target_exist(dst)
        if dst_model is None:
            await anyio.to_thread.run_sync(shutil.copy, self.local_path, dst)
        else:
            # 目标处存在资源
            if overwrite == 'F':
                raise HTTPException(412)  # RFC-4918 9.8.5
            else:
                await dst_model.delete()
                await anyio.to_thread.run_sync(shutil.copy, self.local_path, dst)

    async def content_loader(
            self, rng: Tuple[int, int], chunk_size: int = 64 * 1024
    ) -> AsyncGenerator[bytes, None]:
        """
        异步生成器 用于读取文件内容
        :param rng:
        :param chunk_size:
        :return:
        """
        s, e = rng
        total_length = e - s + 1  # 需要发送的字节数
        async with await anyio.open_file(self.local_path, mode="rb") as file:
            size_sent: int = 0  # 已发送的字节数
            await file.seek(s)
            while size_sent < total_length:
                block_size = min(chunk_size, total_length - size_sent)
                block = await file.read(block_size)
                size_sent += block_size
                yield block

    def _get_range(self, rng_header: Optional[str]) -> Optional[Tuple[int, int, int]]:
        """
        根据Range Header返回(start, end, total_length)
        :param rng_header:
        :return:
        """
        if rng_header is None:
            return None

        total_length = self.size
        rng = rng_header.lower().replace(" ", '').replace("bytes=", '')
        t = rng.split('-')
        try:
            s = int(t[0])
        except Exception:
            s = 0

        try:
            e = int(t[1])
        except Exception:
            e = total_length - 1

        if s >= e:
            s, e = 0, total_length - 1

        send_length = min(e - s + 1, total_length - s)
        return s, e, send_length

    def _make_resp_header(self, rng: Optional[Tuple[int, int, int]]) -> dict:
        headers = dict()
        total_length = self.size
        if rng:
            s, e, send_length = rng
            content_rng = f"bytes {s}-{e}/{total_length}"  # RFC 7233 4.2 Content-Range
            headers["Content-Range"] = content_rng
            headers["Content-length"] = str(send_length)
        else:
            headers["Content-length"] = str(total_length)

        content_type = self.PROP_MAPPING["getcontenttype"]
        if content_type:
            headers["Content-type"] = content_type

        headers["Last-modified"] = self.PROP_MAPPING["getlastmodified"]

        return headers

    def response_for_get(
            self, req_rng: str = None, chunk_size: int = 64 * 1024,
            head_only: bool = False
    ):
        rng_param = self._get_range(req_rng)
        status_code = 206 if rng_param else 200
        headers = self._make_resp_header(rng_param)

        if head_only:
            resp = Response(status_code=status_code, headers=headers)
        elif rng_param and rng_param[-1] <= 0:
            resp = Response(status_code=status_code, headers=headers)
        else:
            # 若未解析出Range Header则设置Range为(0, total_size - 1)
            bytes_rng: Tuple[int, int] = rng_param[:2] if rng_param else (0, self.size - 1)
            loader = self.content_loader(bytes_rng)
            resp = StreamingResponse(content=loader, status_code=status_code, headers=headers)
        return resp


class LocalResourceCreator(ResourceCreator):
    """
    资源创建: 对应MKCOL和PUT方法
    """

    async def make_collection(self):
        is_exists = await anyio.to_thread.run_sync(os.path.exists, self.dst)
        if not is_exists:
            await anyio.to_thread.run_sync(os.mkdir, self.dst)
        else:
            # 目标url已经存在资源
            raise HTTPException(405)

    async def put_file(self, request: Request):
        if await anyio.to_thread.run_sync(os.path.isdir, self.dst):
            # 对已存在的Collection的请求视为错误  RFC-4918 9.7.2
            raise HTTPException(405)

        async with await anyio.open_file(self.dst, mode="wb") as f:
            async for chunk in request.stream():
                await f.write(chunk)


if __name__ == "__main__":
    pass
